package flink_p1

import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.api.common.state.KeyedStateStore
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, WindowedStream}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

object FlinkTest_15_window_reduce {


  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val socketStream: DataStream[String] = env.socketTextStream("127.0.0.1", 8889)


    val windowedStream: WindowedStream[(String, Int), String, TimeWindow] = socketStream
      .map((_, 1))
      .keyBy(_._1)
      .timeWindow(Time.seconds(10))

    // 1. 简单的reduce
    //    windowedStream.reduce(new ReduceFunction[(String, Int)] {
    //      //每产生一条记录就执行一次reduce方法
    //      override def reduce(value1: (String, Int), value2: (String, Int)): (String, Int) = {
    //        (value1._1, value1._2 + value1._2)
    //      }
    //    }).print()


    // 2. 使用WindowFunction
    //    windowedStream.reduce(new ReduceFunction[(String, Int)] {
    //      //每产生一条记录就执行一次reduce方 法
    //        override def reduce(value1: (String, Int), value2: (String, Int)): (String, Int) = {
    //          (value1._1, value1._2 + value1._2)
    //        }
    //      },
    //      // WindowFunction会在每一个窗口计算完毕之后调用一次，可用于结果入库等操作
    //      // 注意导包 import org.apache.flink.streaming.api.scala.function.WindowFunction
    //      new WindowFunction[(String, Int),(String, Int), String, TimeWindow] {
    //      override def apply(key: String, window: TimeWindow, input: Iterable[(String, Int)], out: Collector[(String, Int)]): Unit = {
    //          println("WindowFunction::apply call....")
    //          println(s"key: $key")
    //          println(s"start: ${window.getStart}")
    //          println(s"end: ${window.getEnd}")
    //          println(s"input ${input.toList}")
    //        //发射给下游
    //          out.collect(input.head)
    //
    //      }
    //    }).print()


    //3 processWindowFunction


    windowedStream.reduce(new ReduceFunction[(String, Int)] {
      override def reduce(value1: (String, Int), value2: (String, Int)): (String, Int) = {
        (value1._1, value1._2 + value2._2)

      }
    },
      //ProcessWindowFunction 也是在window计算结束时调用，通过  context可以获取到更多的上下文信息
      new ProcessWindowFunction[(String, Int), (String, Int), String, TimeWindow] {
        override def process(key: String, context: Context, input: Iterable[(String, Int)], out: Collector[(String, Int)]): Unit = {
          println("WindowFunction::apply call....")
          println(s"key: $key")
          val window: TimeWindow = context.window
          println(s"start: ${window.getStart}")
          println(s"end: ${window.getEnd}")

          println(s"input ${input.toList}")

          val windowState: KeyedStateStore = context.windowState
          val globalState: KeyedStateStore = context.globalState

          val currentProcessingTime: Long = context.currentProcessingTime
          val currentWatermark: Long = context.currentWatermark


          //发射给下游
          out.collect(input.head)

        }
      })


    env.execute()
  }

}
